package com.yuan.kafkastudy.consumer;

import com.google.gson.Gson;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import com.yuan.kafkastudy.constant.MsgType;
import com.yuan.kafkastudy.constant.TopicName;
import com.yuan.kafkastudy.domain.*;
import com.yuan.kafkastudy.mapper.*;
import com.yuan.kafkastudy.proto.PlatformProto;
import com.yuan.kafkastudy.proto.protoJava.*;
import com.yuan.kafkastudy.service.impl.*;
import jdk.nashorn.internal.runtime.JSONFunctions;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@Component
public class KafkaConsumer1 {
    @Resource
    private ThreadPoolExecutor executor;
    @Resource
    private SimulationTabServiceImpl simulationService;
    @Resource
    private PlatformTabServiceImpl platformService;
    @Resource
    private SensorTabServiceImpl sensorService;
    @Resource
    private CommTabServiceImpl commService;
    @Resource
    private WeaponTabServiceImpl weaponService;
    @Resource
    private EnvironmentTabServiceImpl environmentService;
    public void runConsumer() throws InvalidProtocolBufferException {
        //创建配置
        Map<String, Object> config = new HashMap();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group18");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //最大拉去记录数
        config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5000);
        //创建消费者
        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer(config);
        //订阅主题
        consumer.subscribe(Collections.singleton(TopicName.TOPIC_NAME_1));
        //拉去数据
        List<SimulationTab> simList = new ArrayList<>();
        List<PlatformTab> platformList = new ArrayList<>();
        List<SensorTab> sensorList = new ArrayList<>();
        List<CommTab> commList = new ArrayList<>();
        List<WeaponTab> weaponList = new ArrayList<>();
        List<EnvironmentTab> environmentList = new ArrayList<>();
        //异步队列
        List<Future> futures = new ArrayList<>();
        int emptyPollTimes = 0;          // 空轮询计数器
        final int MAX_EMPTY_TIMES = 3;   // 连续 3 次没数据就认为 Topic 已空
        boolean keepLive = true;
        Gson gson = new Gson();
        long startTime = System.currentTimeMillis();
        while (keepLive){
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500));
            for (ConsumerRecord<String, byte[]> record : records){
                log.info("拉取数据的数量"+records.count());
                //解析数据,存储再列表中
                //获取数据类型
                byte[] value = record.value();
                byte msgType = value[0];              // 头部
                byte[] protoBytes = Arrays.copyOfRange(value, 1, value.length);  //内容
                //log.info("收到消息类型: {}", msgType);
                switch (msgType){
                    case MsgType.SIMULATION:
                        log.info("sim偏移量"+record.offset());
                        SimProto.Simulation simulation = SimProto.Simulation.parseFrom(protoBytes);
                        String simStr = JsonFormat.printer().preservingProtoFieldNames().print(simulation);
                        SimulationTab simulationTab = gson.fromJson(simStr, SimulationTab.class);
                        simulationTab.setSim_time_unit(null);
                        simList.add(simulationTab);
                        consumer.commitAsync();//异步提交偏移量
                        //提交异步任务
                        if (simList.size() == 10000){
                            List<SimulationTab> newList = new ArrayList<>(simList);
                            simList.clear();
                            CompletableFuture<Void> simfuture = CompletableFuture.runAsync(() -> {
                                boolean saveResult = simulationService.saveBatch(newList);
                            },executor);
                            futures.add(simfuture);
                        }
                        break;
                    case MsgType.PLATFORM:
                        log.info("platform偏移量"+record.offset());
                        long startHandleTime = System.currentTimeMillis();
                        PlatformProto.Platform platform = PlatformProto.Platform.parseFrom(protoBytes);
                        String platformStr = JsonFormat.printer().print(platform);
                        PlatformTab platformTab = gson.fromJson(platformStr, PlatformTab.class);
                        long endHandleTime = System.currentTimeMillis();
                        log.info("platform解析耗时："+(endHandleTime-startHandleTime));
                        platformList.add(platformTab);
                        consumer.commitAsync();//异步提交
                        //提交异步任务
                        if (platformList.size() == 10000){
                            List<PlatformTab> newList = new ArrayList<>(platformList);
                            platformList.clear();
                            CompletableFuture<Void> platfuture = CompletableFuture.runAsync(() -> {
                                boolean saveResult = platformService.saveBatch(newList);
                            },executor);
                            futures.add(platfuture);
                        }
                        break;
                    case MsgType.SENSOR:
                        log.info("sensor偏移量"+record.offset());
                        SensorProto.Sensor sensor = SensorProto.Sensor.parseFrom(protoBytes);
                        String sensorStr = JsonFormat.printer().print(sensor);
                        SensorTab sensorTab = gson.fromJson(sensorStr, SensorTab.class);
                        sensorList.add(sensorTab);
                        consumer.commitAsync();//异步提交
                        //提交异步任务
                        if (sensorList.size() == 10000){
                            List<SensorTab> newList = new ArrayList<>(sensorList);
                            sensorList.clear();
                            CompletableFuture<Void> sensorfuture = CompletableFuture.runAsync(() -> {
                                boolean saveResult = sensorService.saveBatch(newList);
                            },executor);
                            futures.add(sensorfuture);
                        }
                        break;
                    case MsgType.COMM:
                        log.info("comm偏移量"+record.offset());
                        CommProto.Comm comm = CommProto.Comm.parseFrom(protoBytes);
                        String commStr = JsonFormat.printer().print(comm);
                        CommTab commTab = gson.fromJson(commStr, CommTab.class);
                        commList.add(commTab);
                        consumer.commitAsync();//异步提交
                        //提交异步任务
                        if (commList.size() == 10000){
                            List<CommTab> newList = new ArrayList<>(commList);
                            commList.clear();
                            CompletableFuture<Void> commfuture = CompletableFuture.runAsync(() -> {
                                boolean saveResult = commService.saveBatch(newList);
                            },executor);
                            futures.add(commfuture);
                        }
                        break;
                    case MsgType.WEAPON:
                        log.info("weapon偏移量"+record.offset());
                        WeaponProto.Weapon weapon = WeaponProto.Weapon.parseFrom(protoBytes);
                        String weaponStr = JsonFormat.printer().print(weapon);
                        WeaponTab weaponTab = gson.fromJson(weaponStr, WeaponTab.class);
                        weaponList.add(weaponTab);
                        consumer.commitAsync();//异步提交
                        //提交异步任务
                        if (weaponList.size() == 10000){
                            List<WeaponTab> newList = new ArrayList<>(weaponList);
                            weaponList.clear();
                            CompletableFuture<Void> weaponfuture = CompletableFuture.runAsync(() -> {
                                boolean saveResult = weaponService.saveBatch(newList);
                            },executor);
                            futures.add(weaponfuture);
                        }
                        break;
                    case MsgType.ENVIRONMENT:
                        log.info("environment偏移量"+record.offset());
                        EnvironmentProto.Environment environment = EnvironmentProto.Environment.parseFrom(protoBytes);
                        String environmentStr = JsonFormat.printer().print(environment);
                        EnvironmentTab environmentTab = gson.fromJson(environmentStr, EnvironmentTab.class);
                        environmentList.add(environmentTab);
                        consumer.commitAsync();//异步提交
                        //提交异步任务
                        if (environmentList.size() == 10000){
                            List<EnvironmentTab> newList = new ArrayList<>(environmentList);
                            environmentList.clear();
                            CompletableFuture<Void> environmentfuture = CompletableFuture.runAsync(() -> {
                                boolean saveResult = environmentService.saveBatch(newList);
                            },executor);
                            futures.add(environmentfuture);
                        }
                        break;
                    default:
                        break;
                }
            }
            //当连续拉去三次数据为空时，结束入库操作
            if (records.count() == 0){
                emptyPollTimes++;
                if (emptyPollTimes == MAX_EMPTY_TIMES){
                    //将不够10000条数据的列表进行入库
                    CompletableFuture<Void> endSimFuture = CompletableFuture.runAsync(() -> {
                        simulationService.saveBatch(simList);
                    },executor);
                    futures.add(endSimFuture);
                    CompletableFuture<Void> endPlaFuture = CompletableFuture.runAsync(() -> {
                        platformService.saveBatch(platformList);
                    },executor);
                    futures.add(endPlaFuture);
                    CompletableFuture<Void> endSentFuture = CompletableFuture.runAsync(() -> {
                        sensorService.saveBatch(sensorList);
                    },executor);
                    futures.add(endSentFuture);
                    CompletableFuture<Void> endComFuture = CompletableFuture.runAsync(() -> {
                        commService.saveBatch(commList);
                    },executor);
                    futures.add(endComFuture);
                    CompletableFuture<Void> endWeaFuture = CompletableFuture.runAsync(() -> {
                        weaponService.saveBatch(weaponList);
                    },executor);
                    futures.add(endWeaFuture);
                    CompletableFuture<Void> endEnvFuture = CompletableFuture.runAsync(() -> {
                        environmentService.saveBatch(environmentList);
                    },executor);
                    futures.add(endEnvFuture);
                    keepLive = false;
                }
            }
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        long endTime = System.currentTimeMillis();
        System.out.println("耗时:"+(endTime - startTime));

    }
}
